package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SplitFunction2 extends ProcessFunction<String,String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public SplitFunction2(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //解析json数据
        JSONObject jsonObject = JSON.parseObject(value);

        if (null!=jsonObject.get("err")){
            ctx.output(errorTag,value);
            jsonObject.remove("err");
        }
        if (null!=jsonObject.get("start")){
            ctx.output(startTag,jsonObject.toJSONString());
        }else {
            JSONArray displaysJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");

            jsonObject.remove("diaplays");
            jsonObject.remove("actions");

            out.collect(jsonObject.toJSONString());

            if (null!=displaysJsonArray&& !displaysJsonArray.isEmpty()){
                for (int i=0;i<displaysJsonArray.size();i++){
                    JSONObject displayArray = displaysJsonArray.getJSONObject(i);

                    jsonObject.put("display",displayArray);
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");

            if (null!=actionsJsonArray&& !actionsJsonArray.isEmpty()){
                for (int i=0;i<actionsJsonArray.size();i++){
                    JSONObject actionArray = actionsJsonArray.getJSONObject(i);

                    jsonObject.put("action",actionArray);
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
